package pers.zifeng.distributed.lock.zookeeper.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import pers.zifeng.distributed.lock.utils.PropertiesReader;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @author: zf
 * @date: 2021/05/26 14:06:42
 * @version: 1.0.0
 * @description: zookeeper分布式锁类
 */
@Slf4j
public class ZookeeperLockService {

    // zookeeper连接
    private ZooKeeper zkClient;
    // 根节点
    private static final String LOCK_ROOT_PATH = "/DistributeLocks";
    // 当前节点前缀
    private static final String LOCK_NODE_NAME = "Lock_";
    // 当前节点路径
    private String currentLockPath;
    // 当前线程id
    private static final String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];

    // 监控当前节点的上一节点
    private final Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            log.info("上一节点:" + event.getPath() + "释放");
            synchronized (this) {
                notifyAll();
            }

        }
    };

    public ZookeeperLockService() throws IOException {
        Properties properties = PropertiesReader.getProperties("zookeeper.properties");
        zkClient = new ZooKeeper(properties.getProperty("zk.servers"), 10000, event -> {
            if (event.getState() == Watcher.Event.KeeperState.Disconnected) {
                log.error("当前节点：" + event.getPath() + "连接断开！");
            }
        });
    }

    // 创建当前锁节点
    public void createLock() throws KeeperException, InterruptedException {
        // 首先查看是否存在根节点，不存在则创建
        Stat stat = zkClient.exists(LOCK_ROOT_PATH, false);
        if (stat == null) {
            // 这里创建容器节点，当根节点下没有子节点时自动删除
            zkClient.create(LOCK_ROOT_PATH, "rootPath".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);
        }

        // 创建EPHEMERAL_SEQUENTIAL类型节点
        String currentLockPath = zkClient.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME,
                pid.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        log.info("【当前线程:" + pid + "】 节点创建: " + currentLockPath);
        this.currentLockPath = currentLockPath;
    }

    public void attemptLock() throws Exception {
        // 获取Lock所有子节点，按照节点序号排序
        List<String> currentLockPaths;

        currentLockPaths = zkClient.getChildren(LOCK_ROOT_PATH, false);

        // 排序
        Collections.sort(currentLockPaths);

        int index = currentLockPaths.indexOf(currentLockPath.substring(LOCK_ROOT_PATH.length() + 1));

        // 如果currentLockPath是序号最小的节点，则获取锁
        if (index == 0) {
            log.info("【当前线程:" + pid + "】 锁获得, currentLockPath: " + currentLockPath);
        } else if (index < 0) {
            throw new Exception("【当前线程:" + pid + "】 锁不存在, currentLockPath: " + currentLockPath);
        } else {
            // currentLockPath不是序号最小的节点，监控前一个节点
            String preCurrentLockPath = currentLockPaths.get(index - 1);

            synchronized (watcher) {
                Stat stat = zkClient.exists(LOCK_ROOT_PATH + "/" + preCurrentLockPath, watcher);
                // 假如前一个节点不存在了，比如说执行完毕，或者执行节点掉线，重新获取锁
                if (stat != null) { // 阻塞当前线程，直到preCurrentLockPath释放锁，被watcher观察到，notifyAll后，attemptLock
                    log.info(" 等待前锁释放，preCurrentLockPath：" + preCurrentLockPath);

                    // 这里如果同步块不包括监听设置，则存在一种情况就是在进入synchronized执行watcher.wait()之前前锁释放了，直接触发了watcher的notifyAll后在执行watcher.wait()，这样会导致死锁
                    // 解决方案一:将设置监听与等待操作放在同步块中，防止发生死锁，这里使用方案一
                    // 解决方案二:wait方法设置等待时间，防止发生死锁
                    watcher.wait();
                }
            }
            attemptLock();
        }
    }

    // 释放锁
    public void releaseLock() throws KeeperException, InterruptedException {
        zkClient.delete(currentLockPath, -1);
        log.info("【当前线程:" + pid + "】 锁释放：" + currentLockPath);
    }

    // 关闭连接
    public void closeConnect() throws InterruptedException {
        zkClient.close();
    }

}
